package com.util;

import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;

public class KafkaProducerPool {

    private static KafkaProducerPool kafkaProducerPool;
    private final ConcurrentLinkedQueue<KafkaProducer> pool;
    private final Properties properties;

    private KafkaProducerPool(Properties properties, int poolSize) {
        this.pool = new ConcurrentLinkedQueue<>();
        this.properties = properties;
        for (int i = 0; i < poolSize; i++) {
            pool.add(new KafkaProducer(properties));
        }
    }

    public static KafkaProducerPool getInstance(Properties properties, int poolSize) {
        if (kafkaProducerPool == null) {
            synchronized (KafkaProducerPool.class) {
                if (kafkaProducerPool == null) {
                    kafkaProducerPool = new KafkaProducerPool(properties, poolSize);
                }
            }
        }
        return kafkaProducerPool;
    }

    public KafkaProducer getKafkaProducer() {
        KafkaProducer kafkaProducer = pool.poll();
        if (kafkaProducer == null) {
            kafkaProducer = new KafkaProducer(properties);
        }
        return kafkaProducer;
    }

    public void returnKafkaProducer(KafkaProducer kafkaProducer) {
        pool.offer(kafkaProducer);
    }

    public void close() {
        for (KafkaProducer kafkaProducer : pool) {
            kafkaProducer.close();
        }
    }



}
